#include "stdafx.h"
#include "cppflibinc.h"


using namespace cppflib;
using namespace cppflib::os;
using namespace cppflib::collections;
using namespace cppflib::net;
using namespace cppflib::threading;
using namespace cppflib::util;

static u16_t TESTPORT = 1359;
static pcwstr_t chkStr = _S("hello how are you ? 123456 | +-=!@#$%^&*()");


static void __SendDataFunc(Event *pEvent, int sleepTime)
{
   cppmempool::MemPoolWrap localPool;

   // wait until server ready
   pEvent->Wait();

   Thread::Sleep(100);

   // connect to server
   Socket *pSock = new Socket();
   assert(pSock->Create(CFString(""), 0, Socket::Stream));
   assert(pSock->IsValidSock());
   assert(pSock->Connect(CFString("127.0.0.1"), TESTPORT, 2 * 1000));
   assert(pSock->Connect(CFString("127.0.0.1"), TESTPORT, 2 * 1000));  // connect once more will return immediately
   assert(pSock->Connect(CFString("127.0.0.1"), TESTPORT, 2 * 1000));  // connect once more will return immediately 

   Thread::Sleep(sleepTime);
   assert(!pSock->IsClosedByPeer());
   // send data
   CFString *sendStr = new CFString(chkStr);
   char * utf8SendStr = sendStr->ToUtf8String();
   int sendLen = CFString::Strlen(utf8SendStr) + 1;
   assert(pSock->Send((const u8_t*)utf8SendStr, sendLen, 0) == sendLen);
   
   pSock->Close();

   Thread::Sleep(100);
}

static int SendDataThread(ptr_t p)
{
   __SendDataFunc(reinterpret_cast<Event*>(p), 1000);
   return 0;
}

static int SendDataThread_2(ptr_t p)
{
   __SendDataFunc(reinterpret_cast<Event*>(p), 1500);
   return 0;
}

static int SendDataThread_3(ptr_t p)
{
   __SendDataFunc(reinterpret_cast<Event*>(p), 2000);
   return 0;
}

static void TestListenReceive()
{
   cppmempool::MemPoolWrap localPool;

   // start thread to send data
   Thread a((ThreadEntryPoint(&SendDataThread)));
   Event *startSendEvent = new Event();  // a event
   a.Start(reinterpret_cast<ptr_t>(startSendEvent));

   Thread b((ThreadEntryPoint(&SendDataThread_2)));
   Event *startSendEvent2 = new Event();  // a event
   b.Start(reinterpret_cast<ptr_t>(startSendEvent2));

   Thread c((ThreadEntryPoint(&SendDataThread_3)));
   Event *startSendEvent3 = new Event();  // a event
   c.Start(reinterpret_cast<ptr_t>(startSendEvent3));


   // set up socket to listen
   Socket *pServerSock = new Socket();
   assert(pServerSock->Create(CFString("127.0.0.1"), TESTPORT, Socket::Stream));
   assert(pServerSock->IsValidSock());
   assert(pServerSock->Listen(5));
   
   // signal it is ready to receive data
   startSendEvent->Set();
   startSendEvent2->Set();
   startSendEvent3->Set();
   ListCppObjT<Socket> *sockList = new ListCppObjT<Socket>();
   sockList->AddTail(new Socket());
   sockList->AddTail(new Socket());
   sockList->AddTail(new Socket());

   CFString *peerAddr = new CFString("");
   u16_t peerPort = 0;
   for (int i = 0; i < static_cast<int>(sockList->GetCount()); ++i) {
      assert(pServerSock->Accept(*sockList->Get(i)));
      assert(sockList->Get(i)->GetPeerAddress(*peerAddr, peerPort));
      debug::Print(debug::DL_DEBUG, _S("peer addr = %s, peer port = %d\n"), static_cast<pcwstr_t>(*peerAddr), peerPort);
   }

   // receive data
   BinBuf *binBuf = new BinBuf(256);
   int hasRecvData = 0;
   ListCppObjT<Socket> *tempList = new ListCppObjT<Socket>();
   while (hasRecvData < static_cast<int>(sockList->GetCount())) { 

      tempList->Clear();
      for (int i = 0; i < static_cast<int>(sockList->GetCount()); ++i) {
         if (sockList->Get(i)->IsValidSock())    // add only still open socket
            tempList->AddTail(sockList->Get(i));
      }

      debug::Print(debug::DL_DEBUG, _S("num socket waiting = %d\n"), tempList->GetCount());
      int nrSock = Socket::ChkPendingSockets(*tempList, 5000); // wait for available sockets
      debug::Print(debug::DL_DEBUG, _S("num socket available = %d\n"), nrSock);
      assert(nrSock == tempList->GetCount());
      
      for (int i = 0 ; i < static_cast<int>(tempList->GetCount()); ++i)  {
         
         if (tempList->Get(i)->IsClosedByPeer()) {
            debug::Print(_S("%%%% peer closed %%%%\n"));
            tempList->Get(i)->Close(); // close it so that it won't be considered in the next round
         }
         else {
            binBuf->ClearBuf();
            //printf("nrRecv = %d\n", nrRecv);
            int nrRecv = tempList->Get(i)->Receive(binBuf->GetRawBuf(), (int)binBuf->GetLength(), 0);
            assert(nrRecv >= 0);
         
            binBuf->SetValidLength(nrRecv);

            // check received data
            CFString *recvStr = new CFString((const char*)binBuf->GetRawBuf());
            assert(recvStr->Compare(chkStr) == 0);
            //printf("recv str = %s\n", recvStr->ToUtf8String());

            ++hasRecvData;
         }
      }
   }

   // close
   for (int i = 0; i < static_cast<int>(sockList->GetCount()); ++i)
      sockList->Get(i)->Close();
   pServerSock->Close();

   
   startSendEvent->Close();
   startSendEvent2->Close();
   startSendEvent3->Close();
   
   a.Join();
   b.Join();
   c.Join();
   
}

static void TestConnectTimeout(pcwstr_t address, u16_t port)
{
   debug::Print(_S("Testing Connection timeout ...\n"));

   cppmempool::MemPoolWrap localPool;

   const int CONNECT_TIMEOUT = 2 * 1000; 
   Socket *pSock = new Socket();
   assert(pSock->Create(CFString(""), 0, Socket::Stream));
   assert(pSock->IsValidSock());
   assert(pSock->GetSocketType() == Socket::Stream);
   DateTime * pPreConnect = DateTime::Now();
   assert(!pSock->Connect(CFString(address), port, CONNECT_TIMEOUT)); // should be timeout
   DateTime * pAfterConnect = DateTime::Now();
   TimeDuration td = *pAfterConnect - *pPreConnect;
   //assert(td.TotalMilliSeconds() >= CONNECT_TIMEOUT);

   pSock->Close();
}

static void TestGetLocalAddress()
{
   debug::Print(_S("Testing get local address ...\n"));

   cppmempool::MemPoolWrap localPool;

   Socket *pSock = new Socket();
   assert(pSock->Create(CFString("127.0.0.1"), 0, Socket::Dgram));

   CFString localAddr;
   u16_t localPort;

   assert(pSock->GetLocalAddress(localAddr, localPort));
   //debug::Print(debug::DL_DEBUG, _S("localAddr = %s, localPort = %d\n"), static_cast<pcwstr_t>(localAddr), localPort);
}

static volatile int st_round = 0;

static int CancelWaitThread(ptr_t p)
{
   cppmempool::MemPoolWrap localPool;

   ICancelContext * pCancelContext = reinterpret_cast<ICancelContext*>(p);
   cppmempool::Retain(pCancelContext); // make a retain to prevent releasing by its attaching socket

   Thread::Sleep(500);

   debug::Print(_S("Cancel waiting ...\n"));
   bool ret = pCancelContext->CancelWaiting();
//debug::Print(debug::DL_DEBUG, _S("round = %d, ret = %d ...\n"), st_round, ret ? 1 : 0);
   if (st_round == 0)
      assert(ret);
   else 
      assert(!ret);

   cppmempool::Release(pCancelContext);

   return 0;
}

#define NON_EXIST_HOST _S("10.247.1.123")

static void TestCancelWaitingConnect()
{
   debug::Print(debug::DL_INFO, _S("@@@@ Testing cancel connect ...\n"));

   cppmempool::MemPoolWrap localPool;

   Socket * pCommonSocket = new Socket();
   pCommonSocket->Create(CFString(""), 0, Socket::Stream);
   pCommonSocket->SetAttribute(Socket::SA_CANCELWAIT, lang::Boolean(true)); // enable CancelWaiting

   Thread a((ThreadEntryPoint(&CancelWaitThread)));
   a.Start(reinterpret_cast<ptr_t>(pCommonSocket->GetCancelContext()));
   debug::Print(_S("Try connecting -- 1...\n"));
   assert(!pCommonSocket->Connect(CFString(NON_EXIST_HOST), 1111, 600 * 1000));
   debug::Print(_S("Waiting cancelled...\n"));
   a.Join();

   pCommonSocket->Close(); // connect failed, must be closed to start again
   pCommonSocket->Create(CFString(""), 0, Socket::Stream);
   pCommonSocket->SetAttribute(Socket::SA_CANCELWAIT, lang::Boolean(false)); // disable it this time

   st_round = 1;

   Thread b((ThreadEntryPoint(&CancelWaitThread)));
   b.Start(reinterpret_cast<ptr_t>(pCommonSocket->GetCancelContext()));
   debug::Print(_S("Try connecting -- 2...\n"));
   assert(!pCommonSocket->Connect(CFString(NON_EXIST_HOST), 1234, 3 * 1000));
   debug::Print(_S("Waiting cancelled...\n"));
   b.Join();
}

static int ConnectServerThread(ptr_t p)
{ 
   cppmempool::MemPoolWrap localPool;
   Event *pEvent = reinterpret_cast<Event*>(p);
   // wait until server ready
   pEvent->Wait();

   // connect to server
   Socket *pSock = new Socket();
   assert(pSock->Create(CFString(""), 0, Socket::Stream));
   assert(pSock->Connect(CFString("127.0.0.1"), TESTPORT, 2 * 1000));
   
   debug::Print(_S("Connected socket now sleeps without sending data...\n"));
   Thread::Sleep(5 * 1000);  // sleep without sending data

   pSock->Close();

   return 0;
}

static void TestCancelWaitingReceive()
{
   debug::Print(debug::DL_INFO, _S("@@@@ Testing cancel receive ...\n"));

   cppmempool::MemPoolWrap localPool;

   // start thread to connect to server socket
   Thread a((ThreadEntryPoint(&ConnectServerThread)));
   Event *startSendEvent = new Event();  // a event
   a.Start(reinterpret_cast<ptr_t>(startSendEvent));

   Socket * pServerSock = new Socket();
   pServerSock->Create(CFString("127.0.0.1"), TESTPORT, Socket::Stream);
   pServerSock->Listen(5);
   
   // signal it is ready to receive data
   startSendEvent->Set();

   Socket *pEndPointSock = new Socket();
   pServerSock->Accept(*pEndPointSock);
   
   pEndPointSock->SetAttribute(Socket::SA_CANCELWAIT, lang::Boolean(true));

   u8_t buf[10];
   st_round = 0;

   // launch thread to cancel receive
   Thread c((ThreadEntryPoint(&CancelWaitThread)));
   c.Start(reinterpret_cast<ptr_t>(pEndPointSock->GetCancelContext()));
   debug::Print(_S("Receiving data -- 1...\n"));
   pEndPointSock->ReceiveTimeout(600 * 1000, buf, sizeof(buf), 0);
   debug::Print(_S("Waiting cancelled...\n"));

   Thread::Sleep(1000);

   // launch thread to cancel receive again
   Thread d((ThreadEntryPoint(&CancelWaitThread)));
   d.Start(reinterpret_cast<ptr_t>(pEndPointSock->GetCancelContext()));
   debug::Print(_S("Receiving data -- 2...\n"));
   pEndPointSock->ReceiveTimeout(600 * 1000, buf, sizeof(buf), 0);
   debug::Print(_S("Waiting cancelled...\n"));
   
   startSendEvent->Close();

   d.Join();
   c.Join();
   a.Join();
}

static int TestIPCClient(ptr_t p)
{
   cppmempool::MemPoolWrap localPool;
   CFString * serverSockName = reinterpret_cast<CFString *>(p);
   cppmempool::AddToCurPool(serverSockName);
   
   IPCSocket *pClientSocket = new IPCSocket();
   assert(!pClientSocket->IsServer());
   assert(pClientSocket->Connect(*serverSockName, 2000));

   u8_t buf[10] = {'a','b','c','d','e','f','g','h','i','j'};
   u8_t recvBuf[20];
   for (int i = 0; i < 3; ++i) {
      Thread::Sleep(200);
      pClientSocket->Send(buf, sizeof(buf));
      assert(pClientSocket->Receive(2000, recvBuf, 20) == 14);
      for (int j = 0; j < 14; ++j)
         assert(recvBuf[j] == static_cast<u8_t>(j));
   }

   pClientSocket->Close();

   return 0;
}

static void TestIPCSocket()
{
   debug::Print(debug::DL_INFO, _S("Testing IPC Socket ...\n"));

   cppmempool::MemPoolWrap localPool;
   //pcwstr_t serverSockName = _S("myipcsocket");

   IPCSocket *pServSocket = new IPCSocket(CFString::Empty()); // use a random socket name
   pServSocket->Create();
   assert(pServSocket->IsServer());

   CFString *serverSockName = new CFString(pServSocket->GetSocketName());
   cppmempool::UnlinkFromPool(serverSockName);
   debug::Print(debug::DL_INFO, _S("Server socket name = %s\n"), static_cast<pcwstr_t>(*serverSockName));
   
   Thread a((ThreadEntryPoint(&TestIPCClient)));
   a.Start(reinterpret_cast<ptr_t>(serverSockName));

   Thread::Sleep(500);

   assert(pServSocket->Accept(2000));

   u8_t sendBuf[20];
   u8_t recvBuf[20];
   for (int i = 0; i < 3; ++i) {
      assert(pServSocket->Receive(2000, recvBuf, sizeof(recvBuf)) == 10);
      for (int j = 0; j < 10; ++j)
         assert(recvBuf[j] == 'a' + j); 
      for (int j = 0; j < 14; ++j)
         sendBuf[j] = static_cast<u8_t>(j);
      Thread::Sleep(200);
      pServSocket->Send(sendBuf, 14);
   }

   Thread::Sleep(1000);

   a.Join();

   pServSocket->DisconnectClient();
   pServSocket->Close();
   
}

void TestNet(void)
{
   debug::Print(_S("Testing Net ...\n"));

   TestGetLocalAddress();
   TestConnectTimeout(NON_EXIST_HOST, 1111);
   TestConnectTimeout(_S("127.0.0.1"), 8282);
   TestListenReceive();
   TestCancelWaitingConnect();
   TestCancelWaitingReceive();
   TestIPCSocket();
}
